package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp2;
import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class CommentApp2 extends BaseSQLApp2 {
    public static void main(String[] args) {
        new CommentApp2().start(
                10032,
                4,
                Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO);
        Table commentTab = getCommentTab(tEnv);
        tEnv.createTemporaryView("comment_info",commentTab);
        readBaseDic(tEnv);

        Table commentJoin = getCommentJoin(tEnv);
        getTableResult(tEnv);
        commentJoin.insertInto(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO).execute();
    }

    private TableResult getTableResult(StreamTableEnvironment tableEnv) {
        return tableEnv.executeSql("create table " + Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO + "(" +
                "  id STRING,\n" +
                "  user_id STRING,\n" +
                "  nick_name STRING,\n" +
                "  sku_id STRING,\n" +
                "  spu_id STRING,\n" +
                "  order_id STRING,\n" +
                "  appraise_code STRING,\n" +
                "  appraise_name STRING,\n" +
                "  comment_txt STRING,\n" +
                "  create_time STRING" +
                ")"
                + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
    }


    private Table getCommentJoin(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("SELECT \n" +
                "  id ,\n" +
                "  user_id ,\n" +
                "  nick_name ,\n" +
                "  sku_id ,\n" +
                "  spu_id ,\n" +
                "  order_id ,\n" +
                "  appraise ,\n" +
                "  dic_name appraise_name,\n" +
                "  comment_txt ,\n" +
                "  create_time \n" +
                "FROM comment_info AS c\n" +
                "JOIN base_dic FOR SYSTEM_TIME AS OF c.proc_time AS b\n" +
                "ON c.appraise = b.dic_code;");
    }

    private Table getCommentTab(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['nick_name'] nick_name,\n" +
                "  `data`['sku_id'] sku_id,\n" +
                "  `data`['spu_id'] spu_id,\n" +
                "  `data`['order_id'] order_id,\n" +
                "  `data`['appraise'] appraise,\n" +
                "  `data`['comment_txt'] comment_txt,\n" +
                "   proc_time,\n" +
                "  `data`['create_time'] create_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall' and `table` = 'comment_info' and `type`='insert'");
    }


}
